package com.fy.service.server.service.impl;
import com.alibaba.fastjson.JSONObject;
import com.fy.common.constant.CmdConstants;
import com.fy.common.constant.RedisKeyConstant;
import com.fy.common.entity.SkImGroup;
import com.fy.common.enums.BooleanEnum;
import com.fy.common.model.im.AttributeChannelModel;
import com.fy.common.model.im.MessageModel;
import com.fy.common.model.im.UserOnlineStatus;
import com.fy.common.util.TokenUtil;
import com.fy.dao.mapper.SkImGroupMapper;
import com.fy.framework.component.config.TenantContextHolder;
import com.fy.service.server.handler.SocketServerBusiHandler;
import com.fy.service.server.service.ImGroupService;
import com.fy.service.server.service.ImService;
import com.fy.service.service.ImBrokerService;
import com.fy.service.service.MessageService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.Attribute;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.redisson.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;
import java.util.Map;

@Service
public class ImServiceImpl implements ImService {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Value("${spring.rocketmq.consumer.brokerMessageTopic}")
    private String brokerMessageTopic;

    @Autowired
    private MessageService messageService;
    @Autowired
    private RedissonClient redisson;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private SkImGroupMapper skImGroupMapper;

    //单台服务器所有的channel 连接
    public static Map<String, Channel> channelMapping = SocketServerBusiHandler.channelMapping;

    @Value("${spring.cloud.nacos.discovery.ip}")
    private String host;

    @Value("${netty.server.cer.port}")
    private String prot;

    @Autowired
    private ImBrokerService imBrokerService;

    @Autowired
    private ImGroupService imGroupService;
    @Override
    public void route(MessageModel message, ChannelHandlerContext ctx) throws Exception {

        this.datevalidat(message, ctx);
        TenantContextHolder.setTenant(message.getTenantId());

        //业务路由
        switch (message.getCmd()) {
            case CmdConstants.REQUEST_BINDING:
                //绑定的时候做去做 权限验证
                this.tokenValidat(message, ctx);
                this.bindServer(message, ctx);
                break;
            case CmdConstants.REQUEST_HEARTBEAT:
                this.heartCheck(message, ctx);
                break;
            case CmdConstants.REQUEST_SINGLE_CHAT:
                this.chat(message,CmdConstants.RESPONSE_SEND_SUCCESS,CmdConstants.RESPONSE_SINGLE_CHAT, ctx);
                break;
            case CmdConstants.REQUEST_TYPING:
                this.typing(message, ctx);
                break;
            case CmdConstants.REQUEST_READ_DESTROY:
                this.chat(message,CmdConstants.RESPONSE_SEND_SUCCESS,CmdConstants.RESPONSE_READ_DESTROY, ctx);
                break;
            case CmdConstants.REQUEST_RECALL:
                this.chat(message,CmdConstants.RESPONSE_RECALL_SUCCESS,CmdConstants.RESPONSE_RECALL, ctx);
                break;
            case CmdConstants.REQUEST_READ_MESSAGE:
                this.readMessage(message, ctx);
                break;
            case CmdConstants.REQUEST_GROUP_CHAT:
                imGroupService.groupChat(message,CmdConstants.RESPONSE_SEND_SUCCESS,CmdConstants.RESPONSE_GROUP_CHAT, ctx);
                break;
            //消息删除
            case CmdConstants.REQUEST_DELETE_MESSAGE:
                this.chat(message,CmdConstants.RESPONSE_DELETE_SUCCESS,CmdConstants.RESPONSE_DELETE_MESSAGE, ctx);
                break;
            //阅后即焚已读
            case CmdConstants.REQUEST_READ_DESTROY_READ:
                this.readMessage(message, ctx);
                break;
            default:
                logger.error("未找到有效的cmd" + message.getCmd());
                throw new RuntimeException("未找到有效的cmd");
        }
    }

    /**
     * 绑定客户端连接信息到服务器
     *
     * @param message
     * @param channelHandlerContext
     * @throws Exception
     */
    @Override
    public void bindServer(MessageModel message, ChannelHandlerContext channelHandlerContext) throws Exception {
        MessageModel command = new MessageModel();

        Attribute<AttributeChannelModel> attr = channelHandlerContext.channel().attr(SocketServerBusiHandler.NETTY_CHANNEL_KEY);
        AttributeChannelModel attributeChannelModel = attr.get();
        BeanUtils.copyProperties(message, attributeChannelModel);
        //为ChannelHandlerContext 的自定义参数绑定全局map 的ChannelKey
        //从token中解析出用户id
        Long userId = TokenUtil.getUserIdFromEnToken(message.getToken());
        attributeChannelModel.setUserId(userId);
        attributeChannelModel.createChannelKey(attributeChannelModel);
        //创建redis 中的用户在线列表
        UserOnlineStatus onlineStatus = new UserOnlineStatus();
        BeanUtils.copyProperties(message, onlineStatus);
        onlineStatus.setLastOnlineHost(host + ":" + prot);
        onlineStatus.setIsOnline(BooleanEnum.TRUE.getValue());
        //从token中解析出用户id
        onlineStatus.setUserId(userId);
        //把用户的在线情况，连接netty的节点信息等信息放入缓存
        RMapCache<String, JSONObject> mapObj = redisson.getMapCache(message.getTenantId() + "|" + RedisKeyConstant.NIM_TOKEN_PREFIX + userId);
        JSONObject jsonObject = mapObj.get(message.getDeviceCode());
        if(jsonObject != null){
            jsonObject.put("UserOnlineStatus", JSONObject.toJSON(onlineStatus));
            mapObj.put(message.getDeviceCode(), jsonObject);
            //Boolean set = redissonComponet.setMap(RedisKeyConstant.NIM_TOKEN_PREFIX , onlineStatus.getDeviceCode(), onlineStatus.getToken());
            //redissonComponet.set
            //设置缓存过期时间  这里不设置过期 , 登录或者其他逻辑触发的时候查询出全部设备信息 对过期的进行清理一次
            //redisTemplate.expire(Constants.BROADCAST_ROOM + roomId,10, TimeUnit.SECONDS);
            //存入全局的channel管理对象中
            channelMapping.put(attributeChannelModel.getChannelKey(), channelHandlerContext.channel());
            command.setCmd(CmdConstants.RESPONSE_BINDING);
            command.setProtocolType(message.getProtocolType());
            //channelHandlerContext.write(JSONObject.toJSON(command));
            imBrokerService.push(command, channelHandlerContext);
            //调用维护群在线状态的方法
            this.groupOnlineInfo(message,onlineStatus);

        }


        //维护在线人数统计
     //   groupMembers.put

    }


    /**
     * 权限校验
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void tokenValidat(MessageModel messageModel, ChannelHandlerContext ctx) throws Exception {


        //从ChannelHandlerContext 拿出用户的token信息
        String token = messageModel.getToken();
        Long userId = TokenUtil.getUserIdFromEnToken(token);

        RMapCache<String, JSONObject> mapObj = redisson.getMapCache(messageModel.getTenantId() + "|" + RedisKeyConstant.NIM_TOKEN_PREFIX + userId);


        JSONObject jsonObject = mapObj.get(messageModel.getDeviceCode());
        if (jsonObject == null) {
            messageModel.setCmd(CmdConstants.USER_NOT_LOGIN);
            imBrokerService.push(messageModel, ctx);
            logger.error("用户未登录:" + messageModel.getTenantId() + "|" + RedisKeyConstant.NIM_TOKEN_PREFIX + userId + "------" +  messageModel.getDeviceCode() );
            throw new RuntimeException("用户未登录");
        }


    }


    private void datevalidat(MessageModel messageModel, ChannelHandlerContext ctx) throws Exception {
        if (messageModel.getDeviceCode() == null ||
                messageModel.getProtocolType() == null ||
                messageModel.getDeviceType() == null ||
                messageModel.getToken() == null ||
                messageModel.getTenantId() == null) {

            messageModel.setCmd(CmdConstants.DATA_FORMAT_ERROR);
            messageModel.setProtocolType(messageModel.getProtocolType());
            imBrokerService.push(messageModel, ctx);
            throw new RuntimeException("数据格式不正确 " + JSONObject.toJSONString(messageModel));



        }
        //心跳和绑定方法不做绑定验证
        if(!messageModel.getCmd().equals(CmdConstants.REQUEST_HEARTBEAT) && !messageModel.getCmd().equals(CmdConstants.REQUEST_BINDING)) {
            //验证是否绑定到服务器
            Attribute<AttributeChannelModel> attr = ctx.channel().attr(SocketServerBusiHandler.NETTY_CHANNEL_KEY);
            AttributeChannelModel attributeChannelModel = attr.get();
            if (attributeChannelModel.getChannelKey() == null) {
                messageModel.setCmd(CmdConstants.NOT_BINDING);
                imBrokerService.push(messageModel, ctx);
                ctx.channel().close();
                throw new RuntimeException("未绑定到服务器 " + JSONObject.toJSONString(messageModel));

            }

        }



    }


    @Override
    public void heartCheck(MessageModel message, ChannelHandlerContext ctx) throws Exception {

        MessageModel responseModel = new MessageModel();
        responseModel.setCmd(CmdConstants.RESPONSE_HEARTBEAT);
        responseModel.setProtocolType(message.getProtocolType());
        imBrokerService.push(responseModel, ctx);


    }

    @Override
    public void chat(MessageModel message, String fromCmd, String toCmd, ChannelHandlerContext ctx) throws Exception {
        //持久化消息后返回发送方消息发送成功
        //生成服务器的消息id
        Long sendTime = new Date().getTime();
        message.setSendTime(sendTime);
        MessageModel saveModel = new MessageModel();
        BeanUtils.copyProperties(message, saveModel);
        saveModel.setCmd(toCmd);
        messageService.saveSingleMessage(saveModel);
        message.setMessageId(saveModel.getMessageId());
        MessageModel responseModel = new MessageModel();
        responseModel.setCmd(fromCmd);
        responseModel.setMessageId(message.getMessageId());
        responseModel.setProtocolType(message.getProtocolType());
        responseModel.setSendTime(sendTime);
        responseModel.setClientMessageId(message.getClientMessageId());
        responseModel.setFrom(message.getFrom());
        responseModel.setTo(message.getTo());
        responseModel.setToGroup(message.getToGroup());
        //维护会话列表  修改对方的会话
        TenantContextHolder.setTenant(message.getTenantId());
        imBrokerService.updateSingleChatSession(message);

        //通知发送方消息发送成功
        imBrokerService.push(responseModel, ctx);
        //直接把消息放入消息队列 送到路由中心去离线下发
        //Message
        rocketMQTemplate.convertAndSend(brokerMessageTopic, JSONObject.toJSONString(message));


    }

    @Override
    public void typing(MessageModel message, ChannelHandlerContext ctx) throws Exception {

        rocketMQTemplate.convertAndSend(brokerMessageTopic, JSONObject.toJSONString(message));
    }

    @Override
    public void groupOnlineInfo(MessageModel message, UserOnlineStatus onlineStatus) {
        //查询用户有哪些群
        List<SkImGroup> groupList = skImGroupMapper.selectMyNormalGruops(message.getFrom());

        for (SkImGroup skImGroup : groupList) {
            RMapCache<String,JSONObject> groupMembers =  redisson.getMapCache(message.getTenantId() + "|" + RedisKeyConstant.GROUP_MEMBERS + skImGroup.getId());
            JSONObject onlineInfo = groupMembers.get(message.getFrom());
            if(onlineInfo == null){
                onlineInfo = new JSONObject();

            }
            onlineStatus.setToken(null);
            onlineInfo.put(message.getDeviceCode(),onlineStatus);
            groupMembers.putAsync(String.valueOf(message.getFrom()),onlineInfo);
        }


    }

    @Override
    public void groupOfflineInfo(MessageModel message, UserOnlineStatus onlineStatus) {
        //查询用户有哪些群
        List<SkImGroup> groupList = skImGroupMapper.selectMyNormalGruops(message.getFrom());
        for (SkImGroup skImGroup : groupList) {
            RMapCache<String,JSONObject> groupMembers =  redisson.getMapCache(message.getTenantId() + "|" + RedisKeyConstant.GROUP_MEMBERS + skImGroup.getId());
            JSONObject onlineInfo = groupMembers.get(message.getFrom());
            if(onlineInfo == null){
                onlineInfo = new JSONObject();

            }
            onlineStatus.setToken(null);
            onlineInfo.put(message.getDeviceCode(),onlineStatus);
            groupMembers.putAsync(String.valueOf(message.getFrom()),onlineInfo);
        }

    }

    @Override
    public void readMessage(MessageModel message, ChannelHandlerContext ctx) throws Exception {


        //通知发送方消息发送成功
  /*      MessageModel responseModel = new MessageModel();
        responseModel.setCmd(CmdConstants.RESPONSE_READ_MESSAGE);
        responseModel.setMessageId(message.getMessageId());
        responseModel.setProtocolType(message.getProtocolType());
        responseModel.setSendTime(new Date().getTime());
        imBrokerService.push(responseModel, ctx);*/
        rocketMQTemplate.convertAndSend(brokerMessageTopic, JSONObject.toJSONString(message));


    }



}
